"""
Case Type   : gs_retrieve
Case Name   : 传统集群，验证gs_retrieve工具的---dbname参数，参数正确时执行成功，异常时合理报错
Create At   : 2024/10/30
Owner       : lonely-dance
Description :
    1.修改参数
    2.配置白名单
    3.主机建表并插入数据
    4.构造主备同步差异，主机故障
    5.备升主
    6.旧主以主机拉起
    7.对比新旧主数据差异
    8.在旧主上使用gs_retrieve找回数据，设置--dbname参数为空、参数错误、参数值错误；
    9.在旧主上使用gs_retrieve找回数据,设置--dbname参数正确
    10.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.成功
    7.确有数据差异
    8.合理报错
    8.解码成功，但可能解码不全，为合理预期
    9.成功
History     :
"""

import os
import re
import time
import unittest

from yat.test import Node
from yat.test import macro
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

pri_sh = CommonSH('PrimaryDbUser')
node_num = pri_sh.get_node_num()


@unittest.skipIf(3 > node_num, '不满足一主两备环境跳过')
class RowCompressTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.constant = Constant()
        self.std1_sh = CommonSH('Standby1DbUser')
        self.std2_sh = CommonSH('Standby2DbUser')
        self.pri_user_node = Node('PrimaryDbUser')
        self.std1_user_node = Node('Standby1DbUser')
        self.db_name = 'db_gs_retrieve_0002'
        self.tb_name = 't_gs_retrieve_0002'
        self.role = 'u_gs_retrieve_0002'
        self.retrieve_data_path = os.path.join(
            os.path.dirname(macro.DB_ENV_PATH), 'gs_retrieve_0002')
        self.paras = {'autovacuum': 'off', 'synchronous_commit': 'off',
                      'wal_level': 'logical', 'enable_slot_log': 'on',
                      'max_replication_slots': 8}
        self.default_params = {'autovacuum': 'on', 'synchronous_commit': 'on',
                               'wal_level': 'hot_standby',
                               'enable_slot_log': 'off',
                               'max_replication_slots': 8}
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_list = [f'local replication {self.role}  trust',
                            f'host replication {self.role} 127.0.0.1/32 trust',
                            f'host replication {self.role} ::1/128 trust']

    def test_row_compress(self):
        text = '-----step1:修改参数;expect:成功-----'
        self.log.info(text)
        for key, value in self.paras.items():
            res = self.common.show_param(key)
            self.log.info(key)
            if res != value:
                res = pri_sh.execute_gsguc('set',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           f'{key}={value}')
                self.log.info(res)
                self.log.info('步骤2重启')
                self.assertTrue(res)

        text = "-----step2:备份pg_hba.conf, 配置白名单;expect:成功-----"
        self.log.info(text)
        cmd = f"\\cp {self.pg_hba_path} {self.pg_hba_path}_bak.bak"
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        self.assertNotIn('cannot', res, f'执行失败: {text}')

        sql = f"drop role if exists {self.role};" \
              f"create role {self.role} with login password " \
              f"'{macro.COMMON_PASSWD}';" \
              f"alter role {self.role} with replication sysadmin;"
        res = pri_sh.execut_db_sql(sql)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')

        for i in self.pg_hba_list:
            res = pri_sh.execute_gsguc('reload',
                                       self.constant.GSGUC_SUCCESS_MSG,
                                       '', pghba_param=i)
            self.log.info(res)
            self.assertTrue(res)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        res = pri_sh.restart_db_cluster()
        self.log.info(res)
        time.sleep(2)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '-----step3:建表并插入数据;expect:成功-----'
        self.log.info(text)
        sql = f'drop database if exists {self.db_name};' \
              f'create database {self.db_name};'
        res = pri_sh.execut_db_sql(sql)
        self.log.info(res)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS,
                      res, f'执行失败: {text}')
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS,
                      res, f'执行失败: {text}')
        sql = f'''drop table if exists {self.tb_name} cascade;
        create table {self.tb_name}(
            col_integer integer,    
            col_bigint bigint,    
            col_smallint smallint,    
            col_tinyint tinyint,    
            col_serial serial,    
            col_smallserial smallserial,    
            col_bigserial bigserial,    
            col_float float,    
            col_double double precision,    
            col_date date,    
            col_time time without time zone,    
            col_timestamp timestamp without time zone,    
            col_char char(10),    
            col_varchar varchar(20),    
            col_text text,    
            col_blob blob,    
            col_bytea bytea);
        '''
        sql_res = pri_sh.execut_db_sql(sql, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS,
                      sql_res, f'执行失败: {text}')
        sql = f"""
            insert into {self.tb_name} values(generate_series(1,1000),10,5,25,default,
                default,default,1237.127,123456.1234,date '2022-03-22','14:21:21','2022-04-25',
                '测试','测试工程师','西安',empty_blob(),E'\\xDEADBEEF');
        """
        sql_res = pri_sh.execut_db_sql(sql, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('INSERT 0 1000', sql_res, f'执行失败: {text}')

        text = '-----step4:构造主备同步差异，主机故障;expect:成功-----'
        self.log.info(text)
        self.log.info('--stop备机，主机插入数据--')
        res = self.std1_sh.stop_db_instance()
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        res = self.std2_sh.stop_db_instance()
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        time.sleep(2)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        sql = f"""
            insert into {self.tb_name} values(generate_series(1, 10),10,5,25,default,default,
            default,1237.127,123456.1234,date '2022-03-22','14:21:21','2022-04-25','测试',
            '测试工程师','西安',empty_blob(),E'\\xDEADBEEF');
        """
        sql_res = pri_sh.execut_db_sql(sql, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('INSERT 0 10', sql_res, f'执行失败: {text}')
        self.log.info('--停止主机，起备机--')
        res = pri_sh.stop_db_instance()
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        res = self.std1_sh.start_db_instance(mode='standby')
        self.assertIn(self.constant.RESTART_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        res = self.std2_sh.start_db_instance(mode='standby')
        self.assertIn(self.constant.RESTART_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        time.sleep(5)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)

        text = '-----step5:备升主;expect:成功-----'
        self.log.info(text)
        res = self.std1_sh.execute_gsctl('failover',
                                         self.constant.FAILOVER_SUCCESS_MSG)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')
        res = self.std1_sh.exec_refresh_conf()
        self.log.info(res)
        self.assertTrue(res, '执行失败' + text)
        time.sleep(5)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)

        text = '-----step6:旧主以主机拉起;expect:成功-----'
        self.log.info(text)
        res = pri_sh.start_db_instance(mode='primary')
        self.assertIn(self.constant.RESTART_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        time.sleep(5)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)

        text = '-----step7:对比新旧主数据差异;expect:确有数据差异-----'
        self.log.info(text)
        sql = [f'select count(*) from {self.tb_name};',
               f'select checksum({self.tb_name}::text) from {self.tb_name};']
        self.log.info(sql)
        new_pri_res = [
            self.std1_sh.execut_db_sql(
                i, dbname=self.db_name).splitlines()[-2].strip() for i in sql]
        self.log.info(new_pri_res)
        old_pri_res = [
            pri_sh.execut_db_sql(
                i, dbname=self.db_name).splitlines()[-2].strip() for i in sql]
        self.log.info(old_pri_res)
        self.assertNotEqual(new_pri_res, old_pri_res, f'执行失败: {text}')

        text = '-----step8:在旧主上使用gs_retrieve找回数据，设置--dbname参数为空、参数错误、参数值错误；' \
                'expect：合理报错-----'
        self.log.info(text)
        text = '--设置--dbname参数为空--'
        self.log.info(text)
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_retrieve --newhost={self.std1_user_node.db_host} " \
              f"--newport={self.pri_user_node.db_port} " \
              f"--oldhost={self.pri_user_node.db_host} " \
              f"--oldport={self.pri_user_node.db_port} " \
              f"-U {self.role} -W --dbname " \
              f"-f {self.retrieve_data_path}"
        self.log.info(cmd)
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        self.assertIn('too many command-line arguments', res, f'执行失败: {text}')

        text = '--设置--dbname参数错误--'
        self.log.info(text)
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_retrieve --newhost={self.std1_user_node.db_host} " \
              f"--newport={self.pri_user_node.db_port} " \
              f"--oldhost={self.pri_user_node.db_host} " \
              f"--oldport={self.pri_user_node.db_port} " \
              f"-U {self.role} -W --ddbname {self.db_name} " \
              f"-f {self.retrieve_data_path}"
        self.log.info(cmd)
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        self.assertIn('unrecognized option', res, f'执行失败: {text}')

        text = '--设置--dbname参数值错误--'
        self.log.info(text)
        cmd = f"gs_retrieve --newhost={self.std1_user_node.db_host} " \
              f"--newport={self.pri_user_node.db_port} " \
              f"--oldhost={self.pri_user_node.db_host} " \
              f"--oldport={self.pri_user_node.db_port} " \
              f"-U {self.role} -W --dbname {self.db_name}_123 " \
              f"-f {self.retrieve_data_path}"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                      expect <<EOF
                      set timeout 300
                      spawn {cmd}
                      expect "Password:"
                      send "{macro.COMMON_PASSWD}\\n"
                      expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        res = self.pri_user_node.sh(execute_cmd).result()
        self.log.info(res)
        self.assertIn(f'FATAL:  database \"{self.db_name}_123\" does not exist', res, f'执行失败: {text}')

        text = '-----step9:在旧主上使用gs_retrieve找回数据;' \
               'expect:解码成功，但可能加码不全，为合理预期-----'
        self.log.info(text)
        cmd = f"gs_retrieve --newhost={self.std1_user_node.db_host} " \
              f"--newport={self.pri_user_node.db_port} " \
              f"--oldhost={self.pri_user_node.db_host} " \
              f"--oldport={self.pri_user_node.db_port} " \
              f"-U {self.role} -W --dbname {self.db_name} " \
              f"-f {self.retrieve_data_path}"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                      expect <<EOF
                      set timeout 300
                      spawn {cmd}
                      expect "Password:"
                      send "{macro.COMMON_PASSWD}\\n"
                      expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        res = self.pri_user_node.sh(execute_cmd).result()
        self.log.info(res)
        self.assertIn('retrieve data finished', res, f'执行失败: {text}')

        cmd = f'cat {self.retrieve_data_path}'
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        self.assertIsNotNone(res, f'执行失败: {text}')
        self.assertEqual(res.count('测试工程师'), 10, f'执行失败: {text}')

    def tearDown(self):
        text = '-----step9:清理环境; expect:成功-----'
        self.log.info(text)
        time.sleep(5)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)

        self.log.info('--清理环境--')
        sql = f'drop database if exists {self.db_name};' \
              f'drop role if exists {self.role};'
        drop_res = pri_sh.execut_db_sql(sql)
        self.log.info(drop_res)
        drop_res = self.std1_sh.execut_db_sql(sql)
        self.log.info(drop_res)

        self.log.info('--恢复pg_hba.conf--')
        cmd = f"if [ -e {self.pg_hba_path}_bak.bak ]; " \
              f"then \\cp {self.pg_hba_path}_bak.bak {self.pg_hba_path}; " \
              f"rm -rf {self.pg_hba_path}_bak.bak; fi"
        self.log.info(cmd)
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        self.log.info('--恢复集群--')
        stop_ins_res = self.std1_sh.stop_db_instance()
        build_res = self.std1_sh.execute_gsctl(
            'build', self.constant.REBUILD_SUCCESS_MSG, param='-b full')
        self.log.info(build_res)
        res = self.std1_sh.exec_refresh_conf()
        self.log.info(res)
        build_res = self.std2_sh.execute_gsctl(
            'build', self.constant.REBUILD_SUCCESS_MSG, param='-b full')
        self.log.info(build_res)
        res = self.std2_sh.exec_refresh_conf()
        self.log.info(res)
        time.sleep(5)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        self.log.info('--恢复参数--')
        for key, value in self.default_params.items():
            res = self.common.show_param(key)
            self.log.info(key)
            if res != value:
                res = pri_sh.execute_gsguc('set',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           f'{key}={value}')
                self.log.info(res)

        cmd = f"if [ -e {self.retrieve_data_path} ]; " \
              f"then rm -rf {self.retrieve_data_path}; fi"
        self.log.info(cmd)
        res = self.pri_user_node.sh(cmd).result()
        self.log.info(res)
        res = pri_sh.restart_db_cluster()
        self.log.info(res)
        time.sleep(2)
        status = pri_sh.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, drop_res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, drop_res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, stop_ins_res,
                      f'执行失败: {text}')
        self.assertTrue(build_res, f'执行失败: {text}')
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
